package com.gmall.data.engine;

import com.gmall.data.common.source.SourceFactory;
import com.gmall.data.common.utils.Constants;
import com.gmall.data.functions.DeviceKeySelector;
import com.gmall.data.functions.Map2BeanMapFunction;
import com.gmall.data.functions.RuleProcessFunction;
import lombok.val;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author wufengchi
 * @create 2021-04-20 下午11:10
 */
public class RuleEngineV1 {

    public static void main(String[] args) throws Exception{

        val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);

        // 读取kafka中的日志
        val streamSource = env.addSource(SourceFactory.createConsumer(Constants.ODS_APP_LOG_TOPIC()));

        // 将日志转换为LogBean格式的数据
        streamSource.map(new Map2BeanMapFunction())
                // 以device分组
                .keyBy(new DeviceKeySelector())
                // 核心处理逻辑
                .process(new RuleProcessFunction())
                .print();

        env.execute();

    }
}
